
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Map;
import java.util.PriorityQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

	
	

public class ServerFault{
		
		//线程池
		static ExecutorService fixedThreadPool = Executors.newFixedThreadPool(2);  
		static ExecutorService fixedThreadPool2 = Executors.newFixedThreadPool(10);  
		static ExecutorService fixedThreadPool3 = Executors.newFixedThreadPool(10);  
	
		private PriorityQueue<UserInfo> answers;
		private PriorityQueue<UserInfo> acceptedAnswers;
		private volatile Map<Long, UserInfo> map;

	    
	    public ServerFault(){
	        this.answers = new PriorityQueue<UserInfo>(10, new AnswerComparator());
	        this.acceptedAnswers = new PriorityQueue<UserInfo>(10, new AcceptedAnswerComparator());
	        this.map = new ConcurrentHashMap<Long, UserInfo>();
	    }
	    

	    class AnswerComparator implements Comparator<UserInfo>{
			@Override
			public int compare(UserInfo o1, UserInfo o2) {
				return o1.getAnswerCount() - o2.getAnswerCount();
			}
	    	
	    }
	    
	    class AcceptedAnswerComparator implements Comparator<UserInfo>{
			@Override
			public int compare(UserInfo o1, UserInfo o2) {
			
				return o1.getAcceptedAnsCount() - o2.getAcceptedAnsCount();
				
			}
	    	
	    }
	    
	    /**
	     * 读取user文件线程
	     * @author devin
	     */
	    class   ReadUserThread implements Runnable{

			@Override
			public void run() {
			  	try {
					//readUsers("users-short.xml");
					readUsers("F:\\workspace\\jdk8\\src\\users-short.xml");
				} catch (FileNotFoundException e) {
					e.printStackTrace();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
	    	
	    }
	    
	    /**
	     * 读取post 文件线程
	     * @author devin
	     */
	    class ReaderPostThread implements Runnable {
			public void run() {
				try {
					//readPosts("posts-short.xml");
					readPosts("F:\\workspace\\jdk8\\src\\posts-short.xml");
				} catch (FileNotFoundException e) {
					e.printStackTrace();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}

	    public String parseFieldFromLine(String line, String key) {
	        // We're looking for a thing that looks like:
	    	// [key]="[value]"
	    	// as part of a larger String.
	    	// We are given [key], and want to return [value].

	    	// Find the start of the pattern
	    	String keyPattern = key + "=\"";
			int idx = line.indexOf(keyPattern);

			// No match
			if (idx == -1) return null;

			// Find the closing quote at the end of the pattern
			int start = idx + keyPattern.length();

			int end = start;
			while (line.charAt(end) != '"') {
				end++;
			}

			// Extract [value] from the overall String and return it
			return line.substring(start, end);
	    }




	    public void readUsers(String filename) throws FileNotFoundException, IOException {
	    	BufferedReader b = new BufferedReader(
	    			new InputStreamReader(new FileInputStream(filename), Charset.forName("UTF-8")));
	    	String line;
	    	System.out.println("start read user file");
	    	while ((line = b.readLine()) != null) {
	    	/*	if(parseFieldFromLine(line, "Id") == null){
	    			continue;
	    		}
	    		//initialize every user's information UserInfo and store them in our map
	    		long Id = Long.parseLong(parseFieldFromLine(line, "Id"));
	    		String DisplayName = parseFieldFromLine(line, "DisplayName");
	    		UserInfo u = new UserInfo(0, 0, DisplayName);
	    		map.put(Id, u);*/
	    		fixedThreadPool.execute(new HandlerUserLineThread(line));
        
	    	}
	    	System.out.println("while finish");
	    	/*try {
	    		fixedThreadPool2.shutdown();
				fixedThreadPool2.awaitTermination(Long.MAX_VALUE,TimeUnit.DAYS);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}*/
	    	System.out.println("user pool closed");
	    	b.close();
	    }
	    
	    /**
	     * 对读取到的每一行 User 数据处理
	     * @author devin
	     */
	    class HandlerUserLineThread implements Runnable{
	    	private String line;
    		public HandlerUserLineThread() {
		
    		}
    		
    		public HandlerUserLineThread(String line ) {
    			this.line = line;
    		}
	    	
			@Override
			public void run() {
				if(parseFieldFromLine(line, "Id") == null){
	    			return;
	    		}
	    		//initialize every user's information UserInfo and store them in our map
	    		long Id = Long.parseLong(parseFieldFromLine(line, "Id"));
	    		String DisplayName = parseFieldFromLine(line, "DisplayName");
	    		UserInfo u = new UserInfo(0, 0, DisplayName);
	    		map.put(Id, u);
			}
	    	
	    }


	    public void readPosts(String filename) throws FileNotFoundException, IOException {
	
	
	    	ArrayList<Long> bestAnswerId = new ArrayList<Long>();;//use long in case of a id that is larger than 2^31 - 1;
	    									//can change to int if there is no answer id larger than 2^31 - 1 to save space

	    //	HashMap<Long, Long> answerMap = new HashMap<Long, Long>(); //store map of answer post id and owner id
	    	Map<Long, Long> answerMap = new ConcurrentHashMap<Long, Long>(); //store map of answer post id and owner id

	    	BufferedReader b = new BufferedReader(
	    			new InputStreamReader(new FileInputStream(filename), Charset.forName("UTF-8")));
	    	String line;
	    	while ((line = b.readLine()) != null) {
	    		fixedThreadPool3.execute(new HandlerPostLineThread(line,bestAnswerId,answerMap));	
	    	}//while 循环结束
	    	
	    	
	    	//如果不需要等待所有线程执行完，就执行以后的代码，可以把此代码注释
	    	try {
	    		fixedThreadPool3.shutdown();
				fixedThreadPool3.awaitTermination(Long.MAX_VALUE,TimeUnit.DAYS);
				System.out.println("post pool finish");
	    	} catch (InterruptedException e) {
				e.printStackTrace();
			}
   
        for(Long i : bestAnswerId){
        	long userId = answerMap.get(i);
        	if(map.containsKey(userId)){
        		int newAcceAnsCount = map.get(userId).getAcceptedAnsCount() + 1;
        		if(acceptedAnswers.contains(map.get(userId))){
    				UserInfo newInfo = new UserInfo(map.get(userId).getAnswerCount(), newAcceAnsCount,map.get(userId).getName());
    				acceptedAnswers.remove(map.get(userId));
    				acceptedAnswers.add(newInfo);
    				map.get(userId).updateAcceptedAnsCount(newAcceAnsCount);
    			}
        		else{
        			map.get(userId).updateAcceptedAnsCount(newAcceAnsCount);
        			if(acceptedAnswers.size() < 10){
        				acceptedAnswers.add(map.get(userId));
        			}
        			if(answerMap.containsKey(i)){
        				if(acceptedAnswers.peek().getAcceptedAnsCount() < newAcceAnsCount){
        					acceptedAnswers.poll();
        					acceptedAnswers.add(map.get(userId));
        				}
        			}
        		}
        	}
        }  	
    }
	
 
	    /**
	     * 对读取的post 每行文件做处理
	     * @author devin
	     */
	    class HandlerPostLineThread implements Runnable{
	    	private String line;
	    	private ArrayList<Long> bestAnswerId ;
	    	private Map<Long, Long> answerMap;
	        public HandlerPostLineThread(String line,	ArrayList<Long> bestAnswerId ,Map<Long,Long> answerMap) {
	        	this.line = line;
	        	this.bestAnswerId = bestAnswerId;
	        	this.answerMap = answerMap;
	        }
	        
			@Override
			public void run() {
				if(parseFieldFromLine(line, "PostTypeId") == null){
	    			return;
	    		}

	    		String PostTypeId = parseFieldFromLine(line, "PostTypeId");

	    		if(PostTypeId.equals("1")){//handle question post
	    			if(parseFieldFromLine(line, "AcceptedAnswerId")== null){
	    				return;
	    			}
	    		long acceptedAnswerId = Long.parseLong(parseFieldFromLine(line, "AcceptedAnswerId"));
            
	    		if(answerMap.containsKey(acceptedAnswerId)){
            	
	    			long userId = answerMap.get(acceptedAnswerId);
	    			int newAcceAnsCount = map.get(userId).getAcceptedAnsCount() + 1;
    
	    			if(acceptedAnswers.contains(map.get(userId))){//if current post is already in top 10,update it(delete it and insert again with new parameter)
	    				UserInfo newInfo = new UserInfo(map.get(userId).getAnswerCount(), newAcceAnsCount,map.get(userId).getName());
	    				acceptedAnswers.remove(map.get(userId));//delete original one
	    				acceptedAnswers.add(newInfo);//add new one
	    				map.get(userId).updateAcceptedAnsCount(newAcceAnsCount);
	    			}
	    			else{
	    				map.get(userId).updateAcceptedAnsCount(newAcceAnsCount);
	    				if(acceptedAnswers.size() < 10){
	    					acceptedAnswers.add(map.get(userId));
	    				}
	    				else if(acceptedAnswers.peek().getAcceptedAnsCount() < newAcceAnsCount){
	    					acceptedAnswers.poll();
	    					acceptedAnswers.add(map.get(userId));
	    				}
	    			}
	    		}
	    		else{
	    			bestAnswerId.add(acceptedAnswerId);
	    		}
	    		}

	    		if(PostTypeId.equals("2")){//handle answer post

	    			long answerId = Long.parseLong(parseFieldFromLine(line, "Id"));
	    			long ownerUserId = Long.parseLong(parseFieldFromLine(line, "OwnerUserId"));
	    			answerMap.put(answerId, ownerUserId);

	    			if(map.containsKey(ownerUserId)){
	    				//update answer count by 1
    		
	    				if(answers.contains(map.get(ownerUserId))){
    				
	    					int newCount = map.get(ownerUserId).getAnswerCount() + 1;
 
	    					UserInfo newInfo = new UserInfo(newCount, map.get(ownerUserId).getAcceptedAnsCount(),
	    							map.get(ownerUserId).getName());
    				
	    					answers.remove(map.get(ownerUserId));
	    					answers.add(newInfo);
    				
	    					map.get(ownerUserId).updateAnswerCount(newCount);
    				
	    				}
	    				else{
    				
	    					int newCount = map.get(ownerUserId).getAnswerCount() + 1;
	    					map.get(ownerUserId).updateAnswerCount(newCount);
    			//check if it is top ten(current top 10 may not be the final top 10)
        		
	    					if(answers.size() < 10){
	    						answers.add(map.get(ownerUserId));
	    					}
	    					else if(answers.peek().getAnswerCount() < newCount ){
        				
	    						answers.poll();
	    						answers.add(map.get(ownerUserId));
	    						}
	    				}	        			
	    			}
	    		}
	    	
			}
    		
    	}
    
	    
	    
	    
	    public void run() throws FileNotFoundException, IOException {
	    	// Load our data
	    	/*readUsers("users-short.xml");
	    	readPosts("posts-short.xml");*/
	    	
	    	fixedThreadPool.execute(new ReadUserThread());
	    //	fixedThreadPool.execute(new ReaderPostThread());
	    	try {
	    		System.out.println("main read user");
	        	fixedThreadPool.shutdown();
			boolean c = 	fixedThreadPool.awaitTermination(10,TimeUnit.SECONDS);
			System.out.println(c);
				System.out.println("file read finish ");
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
	    	//print results
	    	System.out.println("Top 10 users with the most accepted answers:");
	    	while(!acceptedAnswers.isEmpty()){
	    		System.out.print(acceptedAnswers.peek().getAcceptedAnsCount());
	    		System.out.println(acceptedAnswers.poll().getName());
	    	}
	    	System.out.println();
	    	System.out.println("Top 10 users with the most answers:");
	    	while(!answers.isEmpty()){
	    		System.out.print(answers.peek().getAnswerCount());
	    		System.out.println(answers.poll().getName());
	    	}
	    }
    

	    public static void main(String[] args) throws FileNotFoundException, IOException {
	    	ServerFault s = new ServerFault();
	    	s.run();
	    }

}



